package com.shujia.flink.core;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo04Parallelism {
    public static void main(String[] args) throws Exception {
        // Flink中的并行度设置

        /*
         * 如何设置并行度：
         * 1、通过env.setParallelism()
         * 2、提交任务时可以设置：
         *        命令行提交：-p 参数设置并行度
         *        WEB界面提交时：通过parallelism输入框设置
         * 3、如果在集群中提交，则默认的并行度可以在flink-conf.yaml配置文件中设置
         * 4、每个算子操作后可以设置
         *
         * 优先级：4 > 1 > 2 > 3
         *
         * 思考：
         *  程序提交到集群运行会产生几个Task？需要几个TaskSlot？
         *  TaskSlot的数量是以任务中最大的并行度为准（Flink中的共享Slot资源分配方式）。
         *  Task的数量会受到并行度改变以及Shuffle操作的影响。
         *
         * 并行度设置多少合适？
         *      有无Shuffle操作划分：
         *          有Shuffle：1000~1w条/s/并行度
         *          无Shuffle：1w~10w条/s/并行度
         *
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        DataStreamSource<String> socketDS = env.socketTextStream("master", 8888);// Socket只能有一个线程即一个并行度处理

        SingleOutputStreamOperator<Tuple2<String, Integer>> kvDS = socketDS.map(w -> Tuple2.of(w, 1), Types.TUPLE(Types.STRING, Types.INT))
                .setParallelism(3);

        KeyedStream<Tuple2<String, Integer>, String> keyedDS = kvDS.keyBy(t2 -> t2.f0);

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordCntDS = keyedDS.sum(1).setParallelism(4);

        wordCntDS.print().setParallelism(3);


        env.execute();


    }
}
